package com.bigdata.kafka;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * kafka 数据发送测试
 *
 * @author xin.ding
 * @date 2020/8/10 15:51
 */
public class SendTest {


    private static String bootstrapServer = "10.243.141.133:9092";
    private static String topic = "test2";
    private static int testNum = 2;
    private static final FastDateFormat DATE_FORMAT = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss.SSS");

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, org.apache.kafka.common.serialization.StringSerializer.class.getName());
        Producer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < testNum; i++) {
            String data = mockData(i);
            ProducerRecord<String, String> msg = new ProducerRecord<>(topic, data);
            producer.send(msg);
            System.out.println("Send: " + data);
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        producer.close();
    }

    private static String mockData(int index) {
        return String.format("Time: %s ,message: %s", DATE_FORMAT.format(System.currentTimeMillis()), index + 1);
    }

    private static String mockDataDfzq() {
        return "{\"apiName\":\"Ouos1031\",\"messageId\":\"b66a464070e741f59b06a00a9f5cfa39\",\"body\":{\"account_id\":\"A0102030405\",\"adjust_quantity\":\"300\",\"share_type\":\"1\",\"market_id\":\"101\",\"fund_id\":\"98800040\",\"operation\":\"1\",\"pbu_id\":\"\",\"security_id\":\"60000\",\"operator\":\"\"}}";
    }
}
